package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.functions.KeywordUDTF;
import com.atguigu.gmall.realtime.bean.KeywordStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description: 搜索关键词主题 相关指标的统计.
 * @Author: tiancy
 * @Create: 2021/12/10
 * 需要启动的程序 : 日志采集服务 `rt_logger.sh start`、`BaseLogApp`、zk、kafka、hdfs、clickhouse
 */
public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 设置表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.检查点相关设置
        /*
        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        StateBackend fsStateBackend = new FsStateBackend(
                "hdfs://hadoop102:8020/gmall/flink/checkpoint/ProvinceStatsSqlApp");
        env.setStateBackend(fsStateBackend);
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        */
        //TODO 3.将数据源定义为动态表
        String groupId = "keyword_stats_app";
        String pageViewSourceTopic = "dwd_page_log";

        //TODO 7.注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);


        tableEnv.executeSql("CREATE TABLE page_view (" +
                /** 我们从Kafka dwd_page_log 主题中读取的数据,是一个json字符串格式的数据,其中common对象中,具体的格式是 key-value格式,通过Flink SQL给的数据类型,我们选用MAP格式,其中获取具体属性的值,固定语法`page['item']`*/
                "common MAP<STRING,STRING>, " +
                "page MAP<STRING,STRING>," +
                "ts BIGINT, " +
                /** 将时间戳字段转化为秒,在使用`Flink SQL中定义的函数,转化为`TIMESTAMP`指定格式,这个格式是Flink SQL官方文档指定的格式  */
                "rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss')) ," +
                /** 通过Flink SQl官网上给定的定义水位线的格式`WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '5'` 来指定水位线的策略,这里指定乱序*/
                "WATERMARK FOR  rowtime  AS  rowtime - INTERVAL '2' SECOND) " +
                "WITH (" + MyKafkaUtil.getDDL(pageViewSourceTopic, groupId) + ")");

        //TODO 4.过滤数据,我只要当前页面日志中`page_id` = 'good_list',也就是商品详情页的数据,因为我们输入`关键字`,点击回车,就会跳转到`商品的详情页`.
        Table fullWordView = tableEnv.sqlQuery("select page['item'] fullword ," +
                "rowtime from page_view  " +
                "where page['page_id']='good_list' " +
                "and page['item'] IS NOT NULL ");

        /*
            TODO 5.利用`UDTF`将数据拆分
             这里需要注意一下,如果我们搜索的字符串为 " Apple iPhoneXSMax (A2104) 256GB 深空灰色 移动联通电信4G手机 双卡双待 " . 经过我们自定义的udtf函数后,
             一行数据会变成多行,并且每行数据,需要我们都加上 "时间字段",这种场景的写法,可以参考官方文档
             https://nightlies.apache.org/flink/flink-docs-release-1.12/zh/dev/table/sql/queries.html#joins `Join 表函数 (UDTF)`
             在使用自定义udtf函数时,需要我们提前注册,再使用.SQL的写法,很类似于两张表做笛卡尔积的操作.
         */
        Table keywordView = tableEnv.sqlQuery("select keyword,rowtime  from " + fullWordView + " ," +

                " LATERAL TABLE(ik_analyze(fullword)) as T(keyword)");

       /*
        TableResult execute = keywordView.execute();
        execute.print();
        */


        //TODO 6.聚合计数
        /*
            在使用聚合操作时,需要使用到`自定义的 UDTF函数`,在使用前,需要我们注册函数.注册方式 : TODO 7.注册自定义函数
         */

        //TODO 6.分组、开窗、聚合计算
        Table keywordStatsSearch = tableEnv.sqlQuery("select keyword,count(*) ct, '"
                + GmallConstant.KEYWORD_SEARCH + "' source ," +
                "DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt," + // 窗口开始时间
                "DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt," + // 窗口结束时间
                "UNIX_TIMESTAMP()*1000 ts from   " + keywordView + // 事件处理时间
                /** 在分组时,指定开窗操作`TUMBLE`,也是Flink SQL的固定写法,分组条件和开窗语句中间使用`逗号分割` */
                " GROUP BY TUMBLE(rowtime, INTERVAL '10' SECOND ),keyword");

        //TODO 8.转换为数据流
        DataStream<KeywordStats> keywordStatsProductDataStream =
                tableEnv.<KeywordStats>toAppendStream(keywordStatsSearch, KeywordStats.class);

        keywordStatsProductDataStream.print();
        //TODO 9.写入到ClickHouse
        keywordStatsProductDataStream.addSink(
                ClickHouseUtil.<KeywordStats>getSinkFunction(
                        "insert into keyword_stats_0609(keyword,ct,source,stt,edt,ts)  " + // 关键字、关键字出现次数、数据来源、窗口开始时间、窗口结束时间、事件处理时间.这里制定了插入数据的字段.
                                "values(?,?,?,?,?,?)"));

        env.execute();
    }
}
